package com.eshipenet.shipnet.equipmentextbuilder
import java.util.Properties

import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import play.api.libs.json.{JsObject, Json}

import scala.util.Try

object EquipmentActionJsonStreamProvider {

  def kafkaSource(env: StreamExecutionEnvironment): DataStream[JsObject] = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "kafka-server:9092")
    properties.setProperty("group.id", "equipmentextbuilder.Equipment")
    val der = new JsonKafkaDeserializer()
    val consumer = new FlinkKafkaConsumer[RawKafkaMessage](
      "shipnettestv3.shipnettestv3.equipment",
      der,
      properties
    )
    consumer.setStartFromEarliest()
    val streamTry = env
      .addSource(consumer)
      .map(msg => {
//        println(msg.value)
        Try(Json.parse(msg.value).as[JsObject])
      })
    val stream = streamTry
      .filter(_.isSuccess)
      .map(_.get)
    stream
  }

  def equipmentActionJsonStream(env: StreamExecutionEnvironment): DataStream[JsObject] = {
    env.fromCollection(
      List(
        Json.obj(
          "schema" -> Json.obj(),
          "payload" -> Json.obj(
            "ts_ms" -> 1,
            "op"    -> "i",
            "after" -> Json.obj(
              "id"        -> "1",
              "name"      -> "n1",
              "modelIds"  -> Json.arr("1"),
              "createdAt" -> 1,
              "updatedAt" -> 1,
            ),
          )
        ),
        Json.obj(
          "schema" -> Json.obj(),
          "payload" -> Json.obj(
            "ts_ms" -> 1,
            "op"    -> "i",
            "after" -> Json.obj(
              "id"        -> "2",
              "name"      -> "n1",
              "modelIds"  -> Json.arr("1"),
              "createdAt" -> 1,
              "updatedAt" -> 2,
            ),
          )
        ),
      ))
  }

}
